package org.lds.message.center.plugin.core.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.lds.message.center.common.model.request.GetTemplateAdminRequest;
import org.lds.message.center.plugin.core.remote.RemoteService;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;

import java.util.List;

public class ClearTemplateCacheConsumer extends AbstractKafkaConsumer<GetTemplateAdminRequest> {
    private final RemoteService remoteService;

    public ClearTemplateCacheConsumer(RemoteService remoteService, Class<GetTemplateAdminRequest> clazz) {
        super(clazz);
        this.remoteService = remoteService;
    }

    @KafkaListener(topics = {"${message-center.clear-template-cache.topic}"}, groupId = "${message-center.clear-template-cache.group-id}")
    public void consume(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        super.consume(records, ack);
    }

    @Override
    protected void doConsume(List<GetTemplateAdminRequest> records) {
        for (GetTemplateAdminRequest request : records) {
            remoteService.clearTemplateCache(request.getAppName(), request.getTemplateName());
        }
    }

}
